package com.mq.produce.simple.v2;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @Classname CustomerProduce
 * @Description 发送到指定分区
 * @Version 1.0.0
 * @Date 2025/3/7 9:53
 * @Created by ncs
 */
public class CustomerProduceCallBackPartions {

    static String kafkaServer = "192.168.125.100:9093,192.168.125.100:9094,192.168.125.100:9095";

    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaServer);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);
        for (int i = 0; i < 50; i++) {
            ProducerRecord<String, String> producerRecord = new ProducerRecord<>("first-topic",i%3,"", "first-message-callback" + i);
            kafkaProducer.send(producerRecord, (metadata, exception) -> {
                if (exception != null) {
                    System.out.println("发送消息异步回调失败,异常信息=" + exception);
                }else {
                    System.out.println("发送消息异步回调成功,topic="+ metadata.topic()+", partition="+metadata.partition());
                }
            });
        }
        kafkaProducer.close();
    }
}
